package rxgo

import (
	"context"
	"fmt"
	"reflect"
	"sync"
	"time"
)

type filteringOperater struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (fop filteringOperater) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false
		for x := range in {
			if end {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}
			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if fop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if fop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		if o.flip != nil {
			buffer := (reflect.ValueOf(o.flip))
			if buffer.Kind() != reflect.Slice {
				panic("buffer is not a slice")
			}
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}

		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

//Debounce only emit an item from an Observable if a particular timespan has passed without it emitting another item
func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.timespan = timespan
	o.threading = ThreadingComputing
	o.operator = debounceOperator
	return o
}

var debounceOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	o.count++
	var temp = o.count

	time.Sleep(o.timespan)
	if temp == o.count {
		end = o.sendToFlow(ctx, item.Interface(), out)
	}
	return
}}

//Distinct suppress duplicate items emitted by an Observable
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.m = map[string]bool{}
	o.operator = distinctOperator
	return o
}

var distinctOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	itemStr := fmt.Sprintf("%v", item)
	if _, ok := o.m[itemStr]; !ok {
		o.m[itemStr] = true
		end = o.sendToFlow(ctx, item.Interface(), out)
	}
	return
}}

//ElementAt emit only item n emitted by an Observable
func (parent *Observable) ElementAt(index int) (o *Observable) {
	o = parent.newFilterObservable("elementAt")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.index = index
	o.operator = elementatOperator
	return o
}

var elementatOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if o.count == o.index {
		end = o.sendToFlow(ctx, item.Interface(), out)
	}
	o.count++
	return
}}

//First emit only the first item, or the first item that meets a condition, from an Observable
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.operator = firstOperator
	return o
}

var firstOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	o.sendToFlow(ctx, item.Interface(), out)
	end = true
	return
}}

//IgnoreElements do not emit any items from an Observable but mirror its termination notification
func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("ignoreElements")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.operator = ignoreelementsOperator
	return o
}

var ignoreelementsOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	return
}}

//Last emit only the last item emitted by an Observable
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("xxx")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.operator = lastOperator
	return o
}

var lastOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	o.flip = append([]interface{}{}, item.Interface())
	return
}}

//Sample emit the most recent item emitted by an Observable within periodic time intervals
func (parent *Observable) Sample(sample chan interface{}) (o *Observable) {
	o = parent.newFilterObservable("sample")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.sample = sample
	o.operator = sampleOperator
	return o
}

var sampleOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	var latest = item.Interface()
	var flag = false
	go func() {
		for !flag {
			select {
			case <-ctx.Done():
				flag = true
			case <-o.sample:
				if latest != nil {
					if o.sendToFlow(ctx, latest, out) {
						flag = false
					}
					latest = nil
				}
			}
		}
	}()
	return
}}

//Skip suppress the first n items emitted by an Observable
func (parent *Observable) Skip(n int) (o *Observable) {
	o = parent.newFilterObservable("skip")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.n = n
	o.operator = skipOperator
	return o
}

var skipOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if o.count >= o.n {
		end = o.sendToFlow(ctx, item.Interface(), out)
	}
	o.count++
	return
}}

//SkipLast suppress the last n items emitted by an Observable
func (parent *Observable) SkipLast(n int) (o *Observable) {
	o = parent.newFilterObservable("skipLastx")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.buffer = []interface{}{}
	o.n = n
	o.operator = skiplastOperator
	return o
}

var skiplastOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if o.count == o.n {
		end = o.sendToFlow(ctx, o.buffer[0], out)
		o.buffer = o.buffer[1:]
	} else {
		o.count++
	}
	o.buffer = append(o.buffer, item.Interface())
	return
}}

//Take emit only the first n items emitted by an Observable
func (parent *Observable) Take(n int) (o *Observable) {
	o = parent.newFilterObservable("take")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.n = n
	o.operator = takeOperator
	return o
}

var takeOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if o.count < o.n {
		end = o.sendToFlow(ctx, item.Interface(), out)
	} else {
		end = true
	}
	o.count++
	return
}}

//TakeLast emit only the last n items emitted by an Observable
func (parent *Observable) TakeLast(n int) (o *Observable) {
	o = parent.newFilterObservable("takeLast")
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	o.flip = nil

	o.count = 0
	o.buffer = []interface{}{}
	o.n = n
	o.operator = takelastOperator
	return o
}

var takelastOperator = filteringOperater{opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if o.count == o.n {
		o.buffer = o.buffer[1:]
	} else {
		o.count++
	}
	o.buffer = append(o.buffer, item.Interface())
	o.flip = o.buffer
	return
}}

func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.buf_len = BufferLen
	return o
}
